package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * spark在取数据的时候，顺序是： 内存-》checkpoint（存在hdfs可靠源）=》compute计算
 *
 */
object SparkApiTest06_Cache {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("cache_test")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    context.setCheckpointDir("test_file/ckp")
    val data: RDD[Int] = context.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8))

    val mapData: RDD[(String, Int)] = data.map(item => if (item % 2 == 0) ("A", item) else ("B", item))

    //    mapData.cache()  //其实也是调用了persist(StorageLevel.MEMORY_ONLY)
    //缓存中间数据，只会缓存在task所在节点，只有重复使用的数据才适合缓存
    //    mapData.persist(StorageLevel.MEMORY_AND_DISK)

    //会将数据放在一个比较可靠的地方  checkpoint会在job结束后，另起一个job专门跑一遍此RDD的计算结果数据
    //一般为了调优，都是和persist一起出现 搭配使用，有了persist的缓存，再跑一个job将不需要再那么耗时，直接拿缓存就可以到可靠性的地方
    mapData.checkpoint()


    val groupData: RDD[(String, Iterable[Int])] = mapData.groupByKey()
    groupData.foreach(println)

    val mapValuesData: RDD[(String, Int)] = mapData.mapValues(v => 1)
    val reduceRDD: RDD[(String, Int)] = mapValuesData.reduceByKey(_ + _)
    reduceRDD.foreach(println)
    //spark会将shuffle后的数据自动进行缓存，当二次使用这个结果时，将直接取这个结果进行二次计算
    val newRDD: RDD[(String, String)] = reduceRDD.mapValues(e => e + "gogogo")
    newRDD.foreach(println)

    while (true) {

    }


  }

}
